package com.alison.sparkstream.hook

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.io.File
import scala.sys.ShutdownHookThread

object E1_hook {


  def main(args: Array[String]): Unit = {
    //    stop_hook
    listen_file
  }
  """
    |流式任务需要 7*24 小时执行，但是有时涉及到升级代码需要主动停止程序。如果需要优雅的关闭 Spark Streaming，则在程序中执行如下代码即可，其中 stop 方法参数说明如下：
    |
    |
    |假设我在在 IntelliJ IDEA 中运行程序，程序启动以后，我们点击停止按钮，应用程序会收到中断信号，并触发关闭钩子。
    |并且在关闭过程中尝试优雅地处理当前正在处理的数据，等待已经启动的批处理作业完成。
    |
    |
    |""".stripMargin

  def stop_hook() = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

    // 初始化 StreamingContext，设置微批处理的时间间隔为 5 秒
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // 设置关闭钩子
    ShutdownHookThread {
      // 优雅地关闭资源
      ssc.stop(stopSparkContext = true, stopGracefully = true)
      // 执行其他资源清理操作，例如关闭数据库连接等
      println("程序已经被优雅地关闭了.")
    }

    // 从监控端口读取数据流
    val inputDStream = ssc.socketTextStream("localhost", 9999)
    // 打印结果
    inputDStream.print()
    // 启动 StreamingContext
    ssc.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }

  def listen_file() = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

    // 初始化 StreamingContext，设置微批处理的时间间隔为 5 秒
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // 从监控端口读取数据流
    val inputDStream = ssc.socketTextStream("localhost", 9999)

    // 打印结果
    inputDStream.print()

    // 启动一个线程来监听控制文件
    val directory = this.getClass.getResource("/").toString.replace("file:/", "")
    val fileListenerThread = new FileListenerThread(ssc, directory + "/datas/stop")
    val thread = new Thread(fileListenerThread)
    thread.start()

    // 启动 StreamingContext
    ssc.start()

    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }


  class FileListenerThread(ssc: StreamingContext, controlFilePath: String) extends Runnable {

    private val controlFile = new File(controlFilePath)

    override def run(): Unit = {
      while (!Thread.currentThread().isInterrupted()) {
        Thread.sleep(5_000) // 间隔一段时间检查一次文件
        if (checkControlFile()) {
          // 触发关闭钩子，然后停止 StreamingContext
          ssc.stop(stopSparkContext = true, stopGracefully = true)
          controlFile.delete() // 删除控制文件
          println("程序已经被优雅地关闭了.")
          return
        }
      }
    }

    private def checkControlFile(): Boolean = {
      controlFile.exists() && controlFile.isFile
    }
  }
}
